package com.atguigu.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.bean.TableProcess;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.JdbcUtil;
import com.atguigu.realtime.util.RedisUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;

public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    private Connection phoenixConn;
    private ValueState<Boolean> tableCreateState;
    private Jedis jedis;
    //为了连接外部连接，所以采用RichSinkFunction，而不是SinkFunction
    // invoke 就是自定义方法

    @Override
    public void open(Configuration parameters) throws Exception {
        // open 方法的执行次数，与并行度保持一致
        // 到phonix 的连接 需要在  此 建立
        // 连接对象不支持序列化，不能放在状态里面
        phoenixConn = JdbcUtil.getJdbcConnection(Constant.PHOENIX_DRIVER, Constant.PHOENIX_URL);

        // 为了不 没来一条数据，就建表一次，虽然不真正建表，但是效率会受影响  存入状态中
        tableCreateState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("tableCreateState", Boolean.class));

        // 获取redis连接
        jedis = RedisUtil.getRedisClient();

    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value,
                       Context context) throws Exception {
        // 流中妹来一个元素就执行一次，建立连接需要去open
        // 实现建表和插入数据

        // 1.建表
        checkTable(value);

        // 2. 把数据写入表中
        writeToPhoenix(value);
        // 3.更新缓存
        updateCache(value);

    }

    private void updateCache(Tuple2<JSONObject, TableProcess> value) {
        // 优雅的方法： 当这次操作是update ，并且缓存中存在这条维度数据
        JSONObject data = value.f0;
        TableProcess tp = value.f1;

        String key = tp.getSink_table() + ":" + data.getLong("id");

        if ("update".equals(tp.getOperate_type()) && jedis.exists(key)) {

            // 为了和缓存中的原来的json的key保持大小一致, 务必把json中的key变成大写
            JSONObject upperData = new JSONObject();

            for (Map.Entry<String, Object> entry : data.entrySet()) {
                String upperKey = entry.getKey().toUpperCase();
                Object v = entry.getValue();
                upperData.put(upperKey,v);
            }
            // 更新缓存中的数据
            jedis.setex(key,2*24*60*60,upperData.toJSONString());


            /*jedis.setex(key,2*24*60*60,data.toJSONString()); 没转换大小写*/

        }

        // 粗暴的方法： 把缓存中的数据给删除（不需要转换大小写）
       /* if ("update".equals(tp.getOperate_type()) && jedis.exists(key)) {
            jedis.del(key);
        }*/
    }

    private void writeToPhoenix(Tuple2<JSONObject, TableProcess> value) throws SQLException {
        JSONObject data = value.f0;
        TableProcess tp = value.f1;
        //  执行插入语句  hbase 里没有insert只有upsert
        // upsert  into user(name,id )value(?,?)
        StringBuilder sql = new StringBuilder();
        // 拼接sql语句 TODO
        // into 后面没加空格，导致程序出错，一直在建表！！
        sql
                .append("upsert into ")
                .append(tp.getSink_table())
                .append("(")
                .append(tp.getSink_columns())
                .append(")values(")
                .append(tp.getSink_columns().replaceAll("[^,]+","?"))
                .append(")");

        PreparedStatement ps = phoenixConn.prepareStatement(sql.toString());
        // 给ps 设置占位符 TODO

        String[] columnNames = tp.getSink_columns().split(",");
        for (int i = 0; i < columnNames.length; i++) {
            // 因为有的属性可能不是String 类型，如果直接getString（）可能会报错
            Object v = data.getString(columnNames[i]);
            // 因为有的数据可能是null，会报空指针异常，但是又不能直接转成字符串形式的"null"，后续无法过滤筛选
            ps.setString(i+1, v == null ? null:v.toString());

        }

        ps.execute();
        phoenixConn.commit();
        ps.close();

    }

    private void checkTable(Tuple2<JSONObject, TableProcess> value) throws SQLException, IOException {
        if (tableCreateState.value() == null) {
            TableProcess tp = value.f1;

        // 使用标准的sql 语句 进行建表
        // 1. 拼接建表sql语句   phonix 中必须有主键，是根据hbase的存储特性来的
        // create table if not exists user(id varchar ,name varchar,constraint pk primary key (id,name))
            StringBuilder sql = new StringBuilder();
            sql
                    .append("create table if not exists ")
                    .append(tp.getSink_table())
                    .append("(")
                    .append(tp.getSink_columns().replaceAll("([^,]+)", "$1 varchar"))
                    .append(", constraint pk primary key(")
                    .append(tp.getSink_pk() == null ? "id" : tp.getSink_pk())
                    .append("))")
                    .append(tp.getSink_extend() == null ? "" : tp.getSink_extend());

        System.out.println("建表语句:"+sql.toString());

        //2. 执行sql语句
        PreparedStatement ps = phoenixConn.prepareStatement(sql.toString());
        //2.1 如果sql中有占位符，给占位符赋值
        // 2.2 执行
        ps.execute();
        phoenixConn.commit();
        ps.close();

        // 最后更新表的状态
            tableCreateState.update(true);
        }

    }
        // 在open里面打开的，在close 中关闭
    @Override
    public void close() throws Exception {
        if (phoenixConn != null) {
            phoenixConn.close();
        }
    }
}
